package wordcount;


import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.net.URI;
import java.util.Random;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.StringTokenizer;


import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.lib.map.InverseMapper;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
public class wordcount {
  public static class TokenizerMapper

       extends Mapper<Object, Text, Text, IntWritable>{
    static enum CountersEnum { INPUT_WORDS }
    private final static IntWritable one = new IntWritable(1);
    private Text word = new Text();
    private boolean caseSensitive;
    private Set<String> patternsToSkip = new HashSet<String>();
    private Configuration conf;
    private BufferedReader fis;
    private Set<String> wordstoskip = new HashSet<String>();
    
    public void setup(Context context) throws IOException,
        InterruptedException {
      conf = context.getConfiguration();
      caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
      if (conf.getBoolean("wordcount.skip.patterns", true)) {
        URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
        for (URI patternsURI : patternsURIs) {
          if (patternsURI==patternsURIs[0]){
            Path patternsPath = new Path(patternsURI.getPath());
            String patternsFileName = patternsPath.getName().toString();
            parseSkipFile(patternsFileName);
          }
        }
      }
      if (conf.getBoolean("wordcount.skip.words", true)){
        URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
        for (URI patternsURI : patternsURIs) {
          if (patternsURI!=patternsURIs[0]){
            Path patternsPath = new Path(patternsURI.getPath());
            String patternsFileName = patternsPath.getName().toString();
            wordSkipFile(patternsFileName);
          }
        }
      }
    }
    private void wordSkipFile(String filename){
      try {
        fis = new BufferedReader(new FileReader(filename));
        String pattern = null;
        while ((pattern = fis.readLine()) != null) {
          wordstoskip.add(pattern);
        }
      } catch (IOException ioe) {
        System.err.println("Caught exception while parsing the cached file '"
            + StringUtils.stringifyException(ioe));
      }
    }
    private void parseSkipFile(String fileName) {
      try {
        fis = new BufferedReader(new FileReader(fileName));
        String pattern = null;
        while ((pattern = fis.readLine()) != null) {
          patternsToSkip.add(pattern);
        }
      } catch (IOException ioe) {
        System.err.println("Caught exception while parsing the cached file '"
            + StringUtils.stringifyException(ioe));
      }
    }
    private static boolean isNumber(String str){

      String reg = "^[0-9]+(.[0-9]+)?$";

      return str.matches(reg);

  }
 



    public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
      String line = (caseSensitive) ?
          value.toString() : value.toString().toLowerCase();
      for (String pattern : patternsToSkip) {
        line = line.replaceAll(pattern, "");
      }
      StringTokenizer itr = new StringTokenizer(line);
      while (itr.hasMoreTokens()) {
        word.set(itr.nextToken());
        if (word.getLength()>=3 && !wordstoskip.contains(word.toString()) && !isNumber(word.toString())){
          context.write(word, one);
          Counter counter = context.getCounter(CountersEnum.class.getName(),
              CountersEnum.INPUT_WORDS.toString());
          counter.increment(1);
        }
      }
    }
  }
  public static class IntSumReducer

       extends Reducer<Text,IntWritable,Text,IntWritable> {

    private IntWritable result = new IntWritable();

    public void reduce(Text key, Iterable<IntWritable> values,Context context) throws IOException, InterruptedException {
      int sum = 0;
      for (IntWritable val : values) {
        sum += val.get();

      }
      result.set(sum);
      context.write(key, result);
    }
  }
  public static class AutoReducer

       extends Reducer<IntWritable,Text,Text,NullWritable> {

    private int count = 0;

    public void reduce(IntWritable key, Iterable<Text> values,Context context) throws IOException, InterruptedException {
      for (Text v:values)
      {
        if(count==100)
          return;
        count++;
        String v_string=v.toString();
        String res = count + ":"+ v_string + ", " + key.toString();
        context.write(new Text(res), NullWritable.get()); 

      }
    }
  }
  private static class IntWritableDecreasingComparator extends IntWritable.Comparator {
    public int compare(WritableComparable a, WritableComparable b) {
      return -super.compare(a, b);
    }
    
    public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
        return -super.compare(b1, s1, l1, b2, s2, l2);
    }
}
  public static void main(String[] args) throws Exception {

    Configuration conf = new Configuration();
    GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
    String[] remainingArgs = optionParser.getRemainingArgs();
    if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
      System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
      System.exit(2);
    }
    Path tempDir = new Path("wordcount-temp-" + Integer.toString(  
            new Random().nextInt(Integer.MAX_VALUE)));
    Job job = Job.getInstance(conf, "word count");
    job.setJarByClass(wordcount.class);
    job.setMapperClass(TokenizerMapper.class);
    job.setCombinerClass(IntSumReducer.class);
    job.setReducerClass(IntSumReducer.class);
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(IntWritable.class);
    List<String> otherArgs = new ArrayList<String>();
    for (int i=0; i < remainingArgs.length; ++i) {
      if ("-skip".equals(remainingArgs[i])) {
        job.addCacheFile(new Path(remainingArgs[++i]).toUri());
        job.getConfiguration().setBoolean("wordcount.skip.patterns", true);  //新增属性“wordcount.skip.patterns”,设置为true
      } 
      else if ("-ignore".equals(remainingArgs[i])){
        job.addCacheFile(new Path(remainingArgs[++i]).toUri());
        job.getConfiguration().setBoolean("wordcount.skip.words", true);  //新增属性“wordcount.skip.patterns”,设置为true
      }
      else{
        otherArgs.add(remainingArgs[i]);
      }
    }
    FileInputFormat.addInputPath(job, new Path(otherArgs.get(0)));
    FileOutputFormat.setOutputPath(job, tempDir);
    job.setOutputFormatClass(SequenceFileOutputFormat.class);
    if(job.waitForCompletion(true))
    {
      try{
        Job sortJob = new Job(conf, "sort");
        sortJob.setJarByClass(wordcount.class);
      
        FileInputFormat.addInputPath(sortJob, tempDir);
        sortJob.setInputFormatClass(SequenceFileInputFormat.class);
      /*InverseMapper由hadoop库提供，作用是实现map()之后的数据对的key和value交换*/
        sortJob.setMapperClass(InverseMapper.class);
      /*将 Reducer 的个数限定为1, 最终输出的结果文件就是一个。*/
        sortJob.setNumReduceTasks(1); 
        sortJob.setReducerClass(AutoReducer.class);
        FileOutputFormat.setOutputPath(sortJob, new Path(otherArgs.get(1)));
      
        sortJob.setOutputKeyClass(IntWritable.class);
        sortJob.setOutputValueClass(Text.class);
      /*Hadoop 默认对 IntWritable 按升序排序，而我们需要的是按降序排列。
       * 因此我们实现了一个 IntWritableDecreasingComparator 类,　
       * 并指定使用这个自定义的 Comparator 类对输出结果中的 key (词频)进行排序*/
        sortJob.setSortComparatorClass(IntWritableDecreasingComparator.class);

        System.exit(sortJob.waitForCompletion(true) ? 0 : 1);
      }finally
      {
        FileSystem.get(conf).deleteOnExit(tempDir);
      }
      System.exit(0);
     
    }
    else
    {
      System.exit(1);
    }
  }
}